#pragma once
#include "requestor.hpp"

namespace Json_rpc
{
    namespace client
    {
        // 主题管理模块
        class TopicManager
        {
        public:
            using SubCallback = std::function<void(const std::string &key, const std::string &msg)>;
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager(const Requestor::ptr &requestor) : _requestor(requestor) {}
            bool create(const BaseConnection::ptr &conn, const std::string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_CREATE);
            }
            bool remove(const BaseConnection::ptr &conn, const std::string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_REMOVE);
            }
            bool subscribe(const BaseConnection::ptr &conn, const std::string &key, const SubCallback &cb)
            {
                addSubscribe(key, cb);
                bool ret = commonRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
                if (ret == false)
                {
                    delSubscribe(key);
                    return false;
                }
                return true;
            }
            bool unsubscribe(const BaseConnection::ptr &conn, const std::string &key)
            {
                delSubscribe(key);
                return commonRequest(conn, key, TopicOptype::TOPIC_CANCEL);
            }
            bool publish(const BaseConnection::ptr &conn, const std::string &key, const std::string &msg)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_PUBLISH, msg);
            }
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                TopicOptype type = msg->optype();
                if(type!=TopicOptype::TOPIC_PUBLISH)
                {
                    ELOG("收到错误类型[%d]主题请求操作",(int)type);
                    return ;
                }
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();
                auto callback = getSubscribe(topic_key);
                if(callback==nullptr)
                {
                    ELOG("该主题[%s]未找到处理回调",topic_key.c_str());
                    return ;
                }
                return callback(topic_key,topic_msg);
            }

        private:
            void addSubscribe(const std::string &key, const SubCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callback.insert(std::make_pair(key, cb));
            }
            void delSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callback.erase(key);
            }
            const SubCallback getSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callback.find(key);
                if (it == _topic_callback.end())
                {
                    return SubCallback();
                }
                return it->second;
            }
            bool commonRequest(const BaseConnection::ptr &conn, const std::string &key,
                               TopicOptype type, const std::string &msg = "")
            {
                // 1.构造请求对象，填充数据
                auto req_topic = MessageFactory::create<TopicRequest>();
                req_topic->setId(UuidUtil::uuid());
                req_topic->setMType(MType::REQ_TOPIC);
                req_topic->setOptype(type);
                req_topic->setTopicKey(key);
                if (type == TopicOptype::TOPIC_PUBLISH)
                {
                    req_topic->setTopicMsg(msg);
                }
                // 2.发送请求，等待响应
                BaseMessage::ptr rsp_msg;
                bool ret = _requestor->send(conn, req_topic, rsp_msg);
                if (ret == false)
                {
                    ELOG("主题:[%s] 操作请求失败", key.c_str());
                    return false;
                }
                // 3.判断请求处理是否成功
                TopicResponse::ptr topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(rsp_msg);
                if (!topic_rsp_msg)
                {
                    ELOG("主题操作响应向下转换失败!");
                    return false;
                }
                if (topic_rsp_msg->rCode() != RCode::RCODE_OK)
                {
                    ELOG("主题操作请求出错: %s", errReason(topic_rsp_msg->rCode()).c_str());
                    return false;
                }
                return true;
            }

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, SubCallback> _topic_callback;
            Requestor::ptr _requestor;
        };
    }
}